package com.atguigu.flink.datastream.sink;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author WEIYUNHUI
 * @date 2023/6/14 14:10
 *
 * JdbcSink:
 *   1. 常规写入 sink()
 *   2. exactlyOnceSink() 支持分布式写入 ， 支持分布式事务 ，  支持EOS
 *
 * JDBC步骤:
 *   注册驱动
 *   获取连接
 *   编写SQL
 *   预编译SQL
 *   设置参数
 *   执行SQL
 *   处理结果
 *   关闭连接
 *
 *
 *   写入方式:
 *      1. 直接写入   insert into clicks (xxx) values (xxx)
 *      2. 幂等写入（有主键）
 *            replace into clicks(xxx) values (xxx)
 *            insert into clicks(user ,url ,ts ) values (?,?,?) on Duplicate key update
 */
public class Flink04_JdbcSink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        ds.print("input");

        //将流中的数据写入到Mysql
        SinkFunction<Event> jdbcSink = JdbcSink.<Event>sink(
                //"insert into clicks1 (user ,url ,ts ) values(?,?,?)",
                "replace into clicks1(user, url, ts ) values(?,?,?)" ,
                new JdbcStatementBuilder<Event>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                        preparedStatement.setString(1, event.getUser());
                        preparedStatement.setString(2, event.getUrl());
                        preparedStatement.setLong(3, event.getTs());
                    }
                },

                JdbcExecutionOptions.builder()
                        .withBatchSize(5)  // 批写大小
                        .withBatchIntervalMs(10000)  // 批次间隔时间， 超过指定时间也要写入
                        .withMaxRetries(3)  //重试次数
                        .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUrl("jdbc:mysql://hadoop102:3306/test")
                        .withUsername("root")
                        .withPassword("000000")
                        .build()
        );

        //旧的API
        ds.addSink(jdbcSink) ;

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}


